package org.lql.chaos.kafka.example.serializer;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import org.apache.kafka.common.serialization.Deserializer;
import org.lql.chaos.kafka.example.pojo.Order;
import org.lql.chaos.kafka.example.pojo.OrderItem;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Map;

/**
 * @author: liangqinglong
 * @date: 2025-08-10 03:27
 * @description:
 **/
public class KryoDeserializer<T> implements Deserializer<T> {

	private final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(() -> {
		Kryo kryo = new Kryo();
		kryo.register(ArrayList.class);
		kryo.register(Order.class);
		kryo.register(OrderItem.class);
		kryo.register(BigDecimal.class);
		kryo.register(Timestamp.class);
		kryo.register(String.class);
		return kryo;
	});

	@Override
	public void configure(Map<String, ?> configs, boolean isKey) {
		// 空实现
	}

	@SuppressWarnings("unchecked")
	@Override
	public T deserialize(String topic, byte[] data) {
		if (data == null) return null;
		Kryo kryo = kryoThreadLocal.get();
		try (Input input = new Input(data)) {
			return (T) kryo.readClassAndObject(input);
		}
	}

	@Override
	public void close() {
		// 可选资源释放
	}
}
